package redis.gambing;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.redis.bolt.RedisLookupBolt;
import org.apache.storm.redis.bolt.RedisStoreBolt;
import org.apache.storm.redis.common.config.JedisPoolConfig;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import org.apache.storm.utils.Utils;

import java.util.Map;

import static redis.gambing.GambleFinalString.GAMBLER_ID;

/**
 * 赌博，大小，猜对就结束，计算当前盈余；猜错赌注加倍；如果一直猜错，直到总资产清空
 */
public class GambleTopology {
    public static void main(String[] args) throws Exception {
        // spout 输出赌资，次数，总资产 用三个Hash hash_gambler_1_money:<id, gambling_money>, hash_gambler_2_times:<id, count>, hash_gambler_3_asserts:<id, money>
        // 当存储后，如果赢，该赌徒此生不再赌博，否则继续加倍赌资

        // 创建Redis配置对象
        JedisPoolConfig jedisPoolConfig = new JedisPoolConfig.Builder().setHost("liu").setPort(6379).build();

        // 基本组件实例化
        // 根据指定的id读取赌资，次数，总资产
        // spout 输出赌徒id（GAMBLER_ID）
        GamblerIdSpout gamblerIdSpout = new GamblerIdSpout();
        // 使用三个Hash存储数据，需要写三个类来读取三个Hash的数据，然后发送到下游
        GamblerMoneyInfoLookupMapper gamblerMoneyInfoLookupMapper = new GamblerMoneyInfoLookupMapper();
        RedisLookupBolt gamblerMoneyInfoBolt = new RedisLookupBolt(jedisPoolConfig, gamblerMoneyInfoLookupMapper);
        // filter 过滤money值为0的
        GamblerMoneyInfoFilter gamblerMoneyInfoFilter = new GamblerMoneyInfoFilter();

        GamblerTimesInfoLookupMapper gamblerTimesInfoLookupMapper = new GamblerTimesInfoLookupMapper();
        RedisLookupBolt gamblerTimesInfoBolt = new RedisLookupBolt(jedisPoolConfig, gamblerTimesInfoLookupMapper);
        GamblerAssertsInfoLookupMapper gamblerAssertsInfoLookupMapper = new GamblerAssertsInfoLookupMapper();
        RedisLookupBolt gamblerAssertsInfoBolt = new RedisLookupBolt(jedisPoolConfig, gamblerAssertsInfoLookupMapper);
        // 处理
        // bolt 输出输赢
        GambleResult gambleResult = new GambleResult();
        // bolt 计算总资产，赌博次数
        CalculatedGamblerInfo calculatedGamblerInfo = new CalculatedGamblerInfo();
        // bolt 存储到Redis
        GamblerMoneyInfoStoreMapper gamblerMoneyInfoStoreMapper = new GamblerMoneyInfoStoreMapper();
        RedisStoreBolt gamblerMoneyInfoStoreBolt = new RedisStoreBolt(jedisPoolConfig, gamblerMoneyInfoStoreMapper);
        GamblerTimesInfoStoreMapper gamblerTimesInfoStoreMapper = new GamblerTimesInfoStoreMapper();
        RedisStoreBolt gamblerTimesInfoStoreBolt = new RedisStoreBolt(jedisPoolConfig, gamblerTimesInfoStoreMapper);
        GamblerAssertsInfoStoreMapper gamblerAssertsInfoStoreMapper = new GamblerAssertsInfoStoreMapper();
        RedisStoreBolt gamblerAssertsInfoStoreBolt = new RedisStoreBolt(jedisPoolConfig, gamblerAssertsInfoStoreMapper);

        // 打印
        PrintBolt printBolt = new PrintBolt();

        // 绑定组件
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("gamblerIdSpout", gamblerIdSpout);
        builder.setBolt("gamblerMoneyInfoBolt", gamblerMoneyInfoBolt).fieldsGrouping("gamblerIdSpout", new Fields(GAMBLER_ID));
        builder.setBolt("gamblerMoneyInfoFilter", gamblerMoneyInfoFilter).shuffleGrouping("gamblerMoneyInfoBolt");
        builder.setBolt("gamblerTimesInfoBolt", gamblerTimesInfoBolt).fieldsGrouping("gamblerMoneyInfoFilter", new Fields(GAMBLER_ID));
        builder.setBolt("gamblerAssertsInfoBolt", gamblerAssertsInfoBolt).fieldsGrouping("gamblerTimesInfoBolt", new Fields(GAMBLER_ID));
        builder.setBolt("gambleResult", gambleResult).shuffleGrouping("gamblerAssertsInfoBolt");
        builder.setBolt("calculatedGamblerInfo", calculatedGamblerInfo).shuffleGrouping("gambleResult");
        builder.setBolt("gamblerMoneyInfoStoreBolt", gamblerMoneyInfoStoreBolt).shuffleGrouping("calculatedGamblerInfo");
        builder.setBolt("gamblerTimesInfoStoreBolt", gamblerTimesInfoStoreBolt).shuffleGrouping("calculatedGamblerInfo");
        builder.setBolt("gamblerAssertsInfoStoreBolt", gamblerAssertsInfoStoreBolt).shuffleGrouping("calculatedGamblerInfo");
        builder.setBolt("printBolt", printBolt).shuffleGrouping("calculatedGamblerInfo");

        // 提交拓扑
        Config config = new Config();
        LocalCluster localCluster = new LocalCluster();
        localCluster.submitTopology("gamble-topology", config, builder.createTopology());

    }


    private static class GamblerIdSpout extends BaseRichSpout {
        private SpoutOutputCollector collector;
        private static final String[] gamblerId = {"1001", "1002", "1003", "1004", "1005"};
        private static int index = 0;

        @Override
        public void open(Map<String, Object> conf, TopologyContext context, SpoutOutputCollector collector) {
            this.collector = collector;
        }

        @Override
        public void nextTuple() {
            collector.emit(new Values(gamblerId[index]));
            index = (index + 1) % gamblerId.length;
            Utils.sleep(1000);
        }

        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            declarer.declare(new Fields(GAMBLER_ID));
        }
    }
}
